/*
 * @Author: zhuzhenxxx
 * @Date: 2021-01-28 16:10:38
 * @LastEditTime: 2021-01-28 16:11:14
 * @LastEditors: Please set LastEditors
 * @Description: In User Settings Edit
 * @FilePath: /learn/heartserver.cpp
 */
#include<netinet/in.h>   // sockaddr_in
#include<sys/types.h>    // socket
#include<sys/socket.h>   // socket
#include<arpa/inet.h>
#include<unistd.h>
#include<sys/select.h>   // select
#include<sys/ioctl.h>
#include<sys/time.h>
#include<iostream>
#include<vector>
#include<map>
#include<string.h>
#include<string>
#include<cstdlib>
#include<cstdio>
#include<cstring>

#include <mutex>
#include <thread>
#include <pthread.h>

#include "rtdbFun.h"
#include "fore_rtdb.h"
#include "rtdb_fun.h"
#include "./rapidjson/filereadstream.h"
#include "./rapidjson/document.h"
#include "./rapidjson/rapidjson.h" 
#include "./rapidjson/writer.h"
#include "./rapidjson/stringbuffer.h"
#include "./rapidjson/prettywriter.h"
#include "./rapidjson/filereadstream.h"

#include "./stdafx.h" 
#include "./my_mqtt.h"
#include "./mqtt_msg.h"
#include "./ZNet.h"
#include "./base.h"

#include "./_ccfree.h"
// #include "/home/lilian/jinan/bin/javaserver/src/src_cmu/src_comm/my_mqtt.h"
// #include "/home/lilian/jinan/bin/javaserver/src/include/common/base.h"
// #include "/home/lilian/jinan/bin/javaserver/src/include/net/ZNet.h"

#include "_mysql.h"   // ccfree框架操作MySQL的头文件。

#include "tcpServer.h"


using namespace std;

#define BUFFER_SIZE 1024
typedef struct s_curve
{
	char dev_nodeID[64];
	char name[32];
	int res0;
	char dev_type[32];
	char appName[32];
	char val[289*5];
	int res1;
	int index;
	char res2[64];
	char date[64];           //时间字符串
}s_curve;


using namespace std;

int readJson();
void Createdb();
int qsort_yc_point( const void *a , const void *b );
void db_thread();
void printf_yc_rtdb( );

CLogFile logfile;

Server server(5000);

vector<long> vpthid; //存放线程id的容器

std::deque<s_package> vpackage;
const int DEQUESIZE = 1000;

pthread_mutex_t  dbmutex = PTHREAD_MUTEX_INITIALIZER;


using namespace rapidjson;

int readJson()
{
	FILE* fp = fopen("config.json", "r"); // 非 Windows 平台使用 "r"
	 
	char readBuffer[65536];
	FileReadStream is(fp, readBuffer, sizeof(readBuffer));
	 
	Document root;
	root.ParseStream(is);

	if ( OpenDbms() == false )
	{
		printf("open databases failed.\n");
	}
	else
	{
		printf("open databases success\n");
	}

	char  key[64] = {0};
	char  name[32] = {0};

	float coff;
	char  nodeId[64] = {0};
	char  meaName[32] = {0};
    //判断实时库是否被创建过
	if( !IsLibNoValid(SYS_STRING_KEY_NO) || !IsLibNoValid(FORE_CURVE_NO) )
	{
		Createdb();	
		SaveRtdb();		
		printf("Load rtdb rtdbms\n");
	}
	else {
		if( !LoadRtdb() ) 
		{
			printf(" Load rtdb rtdbms failed2! \n"); 
		}
		else
		{
			printf(" Load rtdb rtdbms sucess2! \n"); 
		} 
	}

	// sleep(2);
	//插入记录（下面有savertdb 所以这里这调用一次就行，即第二次运行程序时，LoadRtdb函数会自动装载上次运行插入的记录，这里需要注意，通常这里插入记录就插入一次就行， ）

	std::string deviceType = root["deviceType"].GetString();
	std::string manufacturerId = root["manufacturerId"].GetString();
	const Value& services = root["services"];
	int ii=1;
	if (services.IsArray())
	{
		for (SizeType i = 0; i < services.Size(); i++) // 使用 SizeType 而不是 size_t
		{
			const Value& properties = services[i]["properties"];
			if( services[i]["properties"].IsArray())
			{
				for (SizeType j = 0; j < properties.Size(); j++)
				{
					std::string name = properties[j]["name"].GetString();
					std::cout << "name is "<< name << std::endl;
					RTDB_YC_CURVE_5MIN   yc_rec;
					sprintf(meaName, "%s", name.c_str() );  //对应xml中的  PhV_phsA、PhV_phsB等
					sprintf(nodeId, "%s_%d:%s", manufacturerId.c_str(), ii,  meaName );
					ii++;
					strcpy(yc_rec.dev_nodeID, nodeId);
					strcpy(yc_rec.name, meaName);
					UpdateRec_stringkey_Ex(FORE_CURVE_NO, nodeId, (char *)&yc_rec, sizeof(RTDB_YC_CURVE_5MIN));
				}
			}
		}
	}
		
	// conn.commit(); // 提交数据库事务。   
	std::cout << deviceType.c_str() << std::endl;
	std::cout << manufacturerId.c_str() << std::endl;

	fclose(fp);
	printf_yc_rtdb();

	return 0;
}


void Createdb()
{
	EmptyAll(); 

	printf("CreateRtdb begin.....\n");
 
	CreateRTdb( "SYS_STRING_KEY_NAME",	SYS_STRING_KEY_NO, SYS_STRING_KEY_COUNT,sizeof(SYS_STRING_KEY));
	 
	CreateRTdb( "FORE_YC_POINT_RTDB_NAME",	FORE_CURVE_NO, FORE_CURVE_COUNT,sizeof(RTDB_YC_CURVE_5MIN)); 
	 	
	printf("CreateRtdb end.....\n");
}


int qsort_yc_point( const void *a , const void *b )
{
	RTDB_YC_CURVE_5MIN * p1 = (RTDB_YC_CURVE_5MIN*)a;
	RTDB_YC_CURVE_5MIN * p2 = (RTDB_YC_CURVE_5MIN*)b; 
 
	if( strcmp( p1->dev_nodeID, p2->dev_nodeID ) > 0 ) 
		return 1;
	else if( strcmp( p1->dev_nodeID, p2->dev_nodeID ) < 0 ) 
		return -1;
	else 
	{
		if( strcmp( p1->name, p2->name ) > 0 ) 
			return 1;
		else if( strcmp( p1->name, p2->name) < 0 ) 
			return -1;
		else return 0;
	}
}
 
void printf_yc_rtdb( )
{
	printf("\n\n -----------------printf_yc_rtdb--------- \n\n");

	tHandle handle = OpenLib(FORE_CURVE_NO);
	if( !IsHandleValid(handle) ) {
		printf("FORE_YC_POINT_NO is InValid \n");
		// return ;
	}
	
	std::vector<RTDB_YC_CURVE_5MIN>  aryYc; 

	RTDB_YC_CURVE_5MIN yc_rec;
	bool ret = ReadFirst(handle, (char*)&yc_rec, sizeof(RTDB_YC_CURVE_5MIN ));

	while(ret)
	{  
		aryYc.push_back(yc_rec);  
		ret = ReadNext(handle, (char*)&yc_rec, sizeof(RTDB_YC_CURVE_5MIN ));
	}
	
	CloseLib( handle );
	
	if( aryYc.size() > 0 )
		qsort(aryYc.data(), aryYc.size(), sizeof(RTDB_YC_CURVE_5MIN), qsort_yc_point );

	for( int i=0; i<aryYc.size(); i++ )
	{
		RTDB_YC_CURVE_5MIN& yc_rec = aryYc[i];
		printf("key = %lld, dev_nodeID = %s, name = %s CurVal0 = %.f CurVal1 = %.f CurVal2 = %.f index = %d val[229]=%.f val[230]=%.f val[231]=%.f\n", \
		yc_rec.KEY, yc_rec.dev_nodeID, yc_rec.name, yc_rec.CurVal0,yc_rec.CurVal1,yc_rec.CurVal2,yc_rec.index,yc_rec.val[229],yc_rec.val[230],yc_rec.val[231]);
		for(int ii = 0; ii < 289; ii++)
		{
			if(ii%10 == 0) printf("\n");
			printf("val[%3d]=%.f ", ii, yc_rec.val[ii]);
		}
		printf("\n\n\n\n");
	}
	 
	printf(" \n \n\n  count = %d  \n", aryYc.size() );
}

void db_thread()
{
	if ( OpenDbms() == false )
	{
		printf("open databases failed.\n");
	}
	else
	{
		printf("open databases success\n");
	}
	if( !LoadRtdb() ) 
	{
		printf(" Load rtdb rtdbms failed2! \n"); 
	}
	else
	{
		printf(" Load rtdb rtdbms sucess2! \n"); 
	} 
	
	char  key[64] = {0};
	char  name[32] = {0};

	float coff;
	char  nodeId[64] = {0};
	char  meaName[32] = {0};
	
	//队列没有值的时候不更新
	s_package dbre;
	RTDB_YC_CURVE_5MIN yc_rec;

	tHandle							handle; 
	float							float_val; 
	short 							curHour;
	short                           curDay;
	short							curMin;
	short							newIndex;
	struct  RTDB_YC_CURVE_5MIN		curve_rec;

	int count =0;

	int ii=0;
	while (true)
	{
		struct tm   * curP;
	
		time_t tt = time(NULL); 
		curP = localtime(&tt);  
		curDay = curP->tm_mday; 
		curHour = curP->tm_hour;
		curMin = curP->tm_min;

		newIndex = curHour * 12 + curMin / 5 ;

		pthread_mutex_lock(&server.mapmutex);
		if(server.v_data.size() == 0 ) 
		{
			pthread_mutex_unlock(&server.mapmutex);
			sleep(2);
			continue;
		}
		else
		{
			dbre = server.v_data.front();
			server.v_data.pop_front();
			pthread_mutex_unlock(&server.mapmutex);
		}

		sprintf(nodeId, "%s", dbre.name);
		strcpy(yc_rec.dev_nodeID ,nodeId);
		FindRec(FORE_CURVE_NO, nodeId,(char*)&yc_rec, sizeof(RTDB_YC_CURVE_5MIN));
		yc_rec.CurVal2 = yc_rec.CurVal1;      
		yc_rec.CurVal1 = yc_rec.CurVal0;           
		yc_rec.CurVal0 = dbre.val;

		yc_rec.val[newIndex] = dbre.val;
		printf("yc_rec.val[newIndex]  is %.f\n",yc_rec.val[newIndex]);

		yc_rec.index = newIndex;

		UpdateRec_stringkey_Ex(FORE_CURVE_NO, nodeId, (char *)&yc_rec, sizeof(RTDB_YC_CURVE_5MIN));

		// 打印插入记录
		sleep(1); 
		printf_yc_rtdb();
	}
}


void save_mysql()
{
	connection conn; // 数据库连接类
    s_curve curve;

	// 登录数据库，返回值：0-成功，其它-失败。
	// 失败代码在conn.m_cda.rc中，失败描述在conn.m_cda.message中。
	if (conn.connecttodb("127.0.0.1,root,lilian,jinan,3306","utf8") != 0)
	{
		printf("connect database failed.\n%s\n",conn.m_cda.message);
	}

	if ( OpenDbms() == false )
	{
		printf("open databases failed.\n");
	}
	else
	{
		printf("open databases success\n");
	}
	if( !LoadRtdb() ) 
	{
		printf(" Load rtdb rtdbms failed2! \n"); 
	}
	else
	{
		printf(" Load rtdb rtdbms sucess2! \n"); 
	} 

	sqlstatement stmt2(&conn); // 操作SQL语句的对象。
	short 							curHour;
	short                           curDay;
	short							curMin;
	short							newIndex;

	time_t _lastlogtime;

	for (;;)
	{
		sleep(5);	
		struct tm lt;
		struct tm nt;

		time_t now = time(0);
		memcpy(&lt, localtime(&_lastlogtime), sizeof(struct tm));
		memcpy(&nt, localtime(&now), sizeof(struct tm));

		if (lt.tm_yday == nt.tm_yday)
		{

		}
		else
		{
			_lastlogtime = now;
			
			tHandle							handle; 
			struct  RTDB_YC_CURVE_5MIN		curve_rec;

			s_package dbre;
			// pthread_mutex_lock(&server.mapmutex);

			handle = OpenLib(FORE_CURVE_NO);
			if( !IsHandleValid(handle) ) return ;
			RTDB_YC_CURVE_5MIN  yc_rec;			
			
			std::vector<RTDB_YC_CURVE_5MIN>  aryYc; 
	
			bool ret = ReadFirst( handle, (char*)&yc_rec, sizeof(RTDB_YC_CURVE_5MIN));
			while( ret )
			{
				
				if( FindRec(FORE_CURVE_NO, curve_rec.dev_nodeID,(char*)&yc_rec, sizeof(RTDB_YC_CURVE_5MIN)) ) 
				{
					aryYc.push_back(yc_rec);  
					ret = ReadNext( handle, (char*)&yc_rec, sizeof(RTDB_YC_CURVE_5MIN));
					continue;
				}

				struct tm* curP;
	
				time_t tt = time(NULL); 
				curP = localtime(&tt);  
				curDay  = curP->tm_mday; 
				curHour = curP->tm_hour;
				curMin  = curP->tm_min;

				newIndex = curHour * 12 + curMin / 5 ;

				memset(&curve,0,sizeof(curve)); // 结构体变量初始化。

				// 为结构体变量的成员赋值。
				// curve.appName = yc_rec.appName;   
				sprintf(curve.dev_nodeID,"%s", yc_rec.dev_nodeID);     
				strcpy(curve.appName , yc_rec.appName);          
				strcpy(curve.name, yc_rec.name);
				curve.res0 = yc_rec.CurVal0;
				strcpy(curve.dev_type,yc_rec.dev_type);
				curve.res1 = yc_rec.CurVal1;    
				// curve.res2 = yc_rec.CurVal2;         
				curve.index = yc_rec.index;

				// if( aryYc.size() > 0 )
				// 	qsort(aryYc.data(), aryYc.size(), sizeof(RTDB_YC_CURVE_5MIN), qsort_yc_point );

				for(int ii = 0; ii < 289; ii++)
				{
					char tmp[5];
					memset(tmp,0,sizeof(tmp));
					sprintf(tmp,"%.f,",yc_rec.val[ii]);
					strcat(curve.val,tmp);
				}
				// for( int i=0; i<aryYc.size(); i++ )
				// {
				// 	RTDB_YC_CURVE_5MIN& rec = aryYc[i];
					
				// 	for(int ii = 0; ii < 289; ii++)
				// 	{
				// 		char tmp[5];
				// 		memset(tmp,0,sizeof(tmp));
				// 		sprintf(tmp,"%.f",rec.val[ii]);
				// 		strcat(curve.val,tmp);
				// 	}
				// }
			
				printf("curve val is %s\n",curve.val);

				char strLocalTime[20];

				memset(strLocalTime,0,sizeof(strLocalTime));
				LocalTime(strLocalTime,"yyyy-mm-dd hh24:mi:ss");
				
				// 准备插入表的SQL语句。
				stmt2.prepare("insert into t_rttable(devnode,name,res0,dev_type,appname,res1,idx,res2,val) \
							values(:1,:2,:3,:4,:5,:6,:7,:8,:9)");
				// prepare方法不需要判断返回值。
				// 为SQL语句绑定输入变量的地址，bindin方法不需要判断返回值。
				stmt2.bindin(1, curve.dev_nodeID,64);
				stmt2.bindin(2, curve.name, 32);
				stmt2.bindin(3, &curve.res0);
				stmt2.bindin(4, curve.dev_type, 32);
				stmt2.bindin(5, curve.appName,32);
				stmt2.bindin(6, &curve.res1);
				stmt2.bindin(7, &curve.index);
				stmt2.bindin(8, curve.res2, 64);
				stmt2.bindin(9, curve.val, 1445);

				printf("strLocalTime is %s\n",strLocalTime);
				stmt2.bindin(10, strLocalTime, 19);


				// 执行SQL语句，一定要判断返回值，0-成功，其它-失败。
				// 失败代码在stmt.m_cda.rc中，失败描述在stmt.m_cda.message中。
				if (stmt2.execute() != 0)
				{
					printf("stmt.execute() failed.\n%s\n%s\n", stmt2.m_sql, stmt2.m_cda.message); 
				}
				else
				{
					printf("成功插入了%ld条记录。\n",stmt2.m_cda.rpc); // stmt.m_cda.rpc是本次执行SQL影响的记录数。
				}
				
				ret = ReadNext( handle, (char*)&yc_rec, sizeof(RTDB_YC_CURVE_5MIN));
			}

			CloseLib(handle);

			conn.commit(); // 提交数据库事务。    
			sleep(5);
			
			_lastlogtime = now;
		}
	}
}

CMyMQTTAsync	        g_myMqttClient;

int myRecv_Msgarrvd(void *context, char *topicName, int topicLen, MQTTAsync_message *message); 

void exit_process( int signo )
{ 
	int pid = getpid();
	printf( "\n syncDbSvr exit signo=%d pid=%d\n", signo, pid );
	
	exit(0);
}


int myRecv_Msgarrvd(void *context, char *topicName, int topicLen, MQTTAsync_message *message)
{ 
	MQTTAsync client = (MQTTAsync)context;
 
	char *  msg_buf = new char[message->payloadlen+1];
	memset( msg_buf, 0, message->payloadlen+1);
	memcpy(msg_buf, message->payload, message->payloadlen); 
 
	printf("Message arrived\n");
	printf("   topic: %s\n", topicName);
	printf("   message payloadlen=%d  : ", message->payloadlen);
  
	printf(" msg_buf = %s \n", msg_buf); 

	delete [] msg_buf;
	msg_buf = NULL; 

	MQTTAsync_freeMessage(&message);
	MQTTAsync_free(topicName);  

	return 1;
}


string  toTimeString(time_t tt, bool IsBjTime )
{
	//time_t lastTime = time(NULL) ; 
	struct tm *p = localtime(&tt);  

	int year  = p->tm_year+1900;
	int month = p->tm_mon+1;
	int day   = p->tm_mday; 
	int Hour   = p->tm_hour; 
	int Minute = p->tm_min;
	int second = p->tm_sec;

	char   chrToken[64] = {0} ;
	if( !IsBjTime ) {
		sprintf(chrToken, "%04d-%02d-%02dT%02d:%02d:%02dZ", p->tm_year+1900, p->tm_mon+1, p->tm_mday, p->tm_hour,p->tm_min, p->tm_sec);
	}
	else {
		sprintf(chrToken, "%04d-%02d-%02d %02d:%02d:%02d", p->tm_year+1900, p->tm_mon+1, p->tm_mday, p->tm_hour,p->tm_min, p->tm_sec);
	}

	return chrToken;
}



bool _send_sub_data(CMyMQTTAsync * pCli,char* dev_type, char* nodeId ,float val)
{ 
	if( pCli == NULL || nodeId == NULL ) return false;

	int pos = 0;
	// int COUNT = rtData.size();
	int SEND_MAX_REC = 50;  
	int idx = 0;
	
	char  topic[128] = {0};
	sprintf(topic, "%s/terminal/dataReport/%s/%s", "dataCenterApp", \
			dev_type, nodeId);

	int  token = 10000;
 
	token = getpid();

	int updateTime = 0;

 
	rapidjson::StringBuffer s;
	rapidjson::Writer<rapidjson::StringBuffer> writer(s);
 	
	char  tmp[32] = {0}; 

	writer.StartObject();

	writer.Key("body");
	writer.StartObject();

	writer.Key("nodeId");
	sprintf(tmp, "%s", nodeId );
	writer.String(tmp);  

	writer.Key("val");
	sprintf(tmp, "%.2f", val );
	writer.String(tmp);  

	// writer.Key("serviceProperties"); 
	// writer.StartObject(); 

	// //dataProperties 
	// writer.Key("dataProperties"); 
	// writer.StartObject(); 

	int  dataTime = 0;
	bool bb = false;
	bool IsHavePowerEvent = false;
	int  iEventVal = -1;

	writer.EndObject();

	writer.Key("token");
	sprintf(tmp, "%d", token );
	writer.String(tmp);
	writer.Key("timestamp"); 

	char tm[64];
	LocalTime(tm);
	writer.String( tm );

	writer.EndObject(); 


	printf( "send payload: %s\n", s.GetString());
	pCli->send_string(1, topic,  (char*)s.GetString() );
	 
	return true;
}

#include "./mqtt_cli.h"
using namespace TDbFun;

void mqtt_thread()
{
	char mqttSvrIp[32] = "tcp://127.0.0.1:1883";   
	char  mqtt_client_id[64] ={0};
	sprintf(mqtt_client_id, "testClientID_%d",  getpid() );

	while(1)
	{
		if( g_myMqttClient.connect_mqtt(mqtt_client_id, mqttSvrIp, myRecv_Msgarrvd) )
		{
			printf("connect_mqtt  ok !\n"); 
			break; 
		}  
	} 

	sleep(1);
	std::vector<FORE_DEV_TYPE>   aryDevType;
	TDbFun::GetDevType(aryDevType); 

	while(1)
	{
		char  key[64]={0};
		char  nodeId[32] ={0};
		char  meaName[32]={0};

		for(int i = 0; i < aryDevType.size(); ++i)
		{
			pthread_mutex_lock(&server.mapmutex);
			for (auto it = server.m_data.begin(); it != server.m_data.end(); it++)
			{
				FORE_DEV_TYPE & devTypeInfo = aryDevType[i];
				char nodeId[64] ={0};
				snprintf(nodeId, "%s" , it->first.c_str(), 64);
				_send_sub_data(&g_myMqttClient,devTypeInfo.deviceType ,nodeId,it->second);
			}
			pthread_mutex_unlock(&server.mapmutex);
		}
		sleep(1);
	}
}

volatile  int   on_connect_ok = 0;

int Pub_msgarrvd(void* context, char* topicName, int topicLen, MQTTAsync_message* m); 
void  pubtopic_loop_check(void* context);

// thread_return_type  process_pubMsg(void* n);
void  process_pubMsg(void* n);

CThreadLock		 g_threadLock; 

CMQTTAsyncPubCli::CMQTTAsyncPubCli()
{  
	//g_repubtopic = false; 
	m_client = NULL;
}

CMQTTAsyncPubCli::~CMQTTAsyncPubCli()
{
	disconnect();
}

void pub_connlost(void *context, char *cause)
{
	base::my_printf(  "\n pub Connection lost \n" );  
}

void onPubDisconnect(void* context, MQTTAsync_successData* response)
{
	base::my_printf("Successful disconnection\n"); 
}

void onPubConnectFailure(void* context, MQTTAsync_failureData* response)
{
	base::my_printf("Connect failed, rc %d\n", response ? response->code : 0); 
} 

void onPubConnect(void* context, MQTTAsync_successData* response)
{
	on_connect_ok = 1;  
}

void onPubSend(void* context, MQTTAsync_successData* response)
{
	 
}

bool CMQTTAsyncPubCli::re_connect_mqtt()
{  
	MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
	  
	MQTTAsync_create(&m_client, m_mqtt_svr_addr.c_str(), PUB_CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
	 
	MQTTAsync_setCallbacks(m_client, this,  pub_connlost, Pub_msgarrvd, NULL);
	
	int rc;

	conn_opts.keepAliveInterval = 20;
	conn_opts.cleansession = 1;
	conn_opts.onSuccess = onPubConnect;
	conn_opts.onFailure = onPubConnectFailure;
	conn_opts.context = m_client;
	if ((rc = MQTTAsync_connect(m_client, &conn_opts)) != MQTTASYNC_SUCCESS)
	{
		base::my_printf("Failed to start connect, return code %d\n", rc);
		return false;
	} 
	
	int   wait_seconds = 6; 
	while ( !on_connect_ok )
	{ 
		if(wait_seconds-- <= 0) break;
		
		base::MySleep(1000L); 
	}
	 
	if( !on_connect_ok ) {
		base::my_printf("MQTTAsync_connect Failed \n");
		return false;
	}

	base::my_printf("pub_topic: connect success \n" );
 
	create_devTypePubtopic_thread();
	 
	base::MySleep(1000*5); 
	pubtopic_loop_check((void*)this);
 	// base::Thread_start(pubtopic_loop_check, (void*)this); 

	return true;
}
 

bool CMQTTAsyncPubCli::ReStartPubThreasd( )
{  
	// CloseThreadHandle();	  
	base::MySleep(1000L); 

	create_devTypePubtopic_thread();
	 

  	return true;
	 
}

bool CMQTTAsyncPubCli::connect_mqtt(char * mqtt_svr_addr)
{ 
	if( strlen( mqtt_svr_addr ) == 0 ) return false;

	m_mqtt_svr_addr = mqtt_svr_addr;
	on_connect_ok = 0;

	disconnect();

	re_connect_mqtt(); 

	return true;
}


void  pubtopic_loop_check(void* context)
{
	CMQTTAsyncPubCli  * pPubCli = (CMQTTAsyncPubCli*)context;
	base::my_printf("---------------------------------create check_devtype_loop---------------------------------- \n" );
	 
	while(1)
	{
		std::vector<FORE_DEV_TYPE>   aryDevType;
		TDbFun::GetDevType(aryDevType); 
		
		base::my_printf("---------------------------------create check_devtype_loop size=%d \n", pPubCli->m_aryDevTypeThreadHandle.size() );

		for( int i=0; i<aryDevType.size(); i++ )
		{
			printf("ary dev type id = %d \n", i);
			if( aryDevType[i].profilename == NULL  ) continue;

			pPubCli->create_devTypePubtopic_thread( aryDevType[i].dev_type );
		} 

		base::MySleep(1000 * 20);
	}  
	return NULL;
}

void CMQTTAsyncPubCli::disconnect()
{ 
	base::my_printf("pub disconnect \n");

	if( MQTTAsync_isConnected( m_client ) ) 
	{ 
		MQTTAsync_disconnectOptions opts = MQTTAsync_disconnectOptions_initializer;
		int rc;  
		 
		opts.onSuccess = onPubDisconnect;
		opts.context = m_client;

		if ((rc = MQTTAsync_disconnect(m_client, &opts)) != MQTTASYNC_SUCCESS)
		{
			base::my_printf("Failed to MQTTAsync_disconnect, return code %d\n", rc);
			exit(EXIT_FAILURE);
		} 
	
		MQTTAsync_destroy(&m_client);
	}
}	 

void CMQTTAsyncPubCli::create_devTypePubtopic_thread(char * devType)
{   
	if( devType == NULL )
	{
		base::my_printf(" create_devTypePubtopic_thread:  load all pubtopic \n"  );
  
		std::vector<FORE_DEV_TYPE>   aryDevType;
		TDbFun::GetDevType( aryDevType ); 
 
		base::my_printf( " devType.size=%d  \n", aryDevType.size() );

		 
		int ii = 0;
		for (int i=0; i<aryDevType.size(); i++) {  
			
			FORE_DEV_TYPE & devTypeInfo = aryDevType[i];

			// if( devTypeInfo.protocolType == NULL || devTypeInfo.manufacturerId == NULL ) continue;

			struct pub_thread_parms * parms = new struct pub_thread_parms; // { &client, QOS, ii++, ary, (char*)strTopic.c_str() };
			parms->c = m_client;
			parms->qos = QOS;
			parms->index = ii++;
			parms->devTypeInfo = devTypeInfo;
			parms->pPubCli = this;
			  
			// pthread_t threadId = base::Thread_start(process_pubMsg, (void*)parms);
			process_pubMsg((void*)parms);
  
			base::MySleep(500L);
		}  
	}
	else 
	{
		FORE_DEV_TYPE * pRec = (FORE_DEV_TYPE*)FindRec_stringkey(FORE_DEV_TYPE_NO, devType);
		// if( pRec == NULL) return  ;

		// if( pRec->protocolType == NULL || pRec->manufacturerId == NULL ) return;
		 
		struct pub_thread_parms * parms = new struct pub_thread_parms; 
		parms->c = m_client;
		parms->qos = QOS;
		parms->index = m_aryDevTypeThreadHandle.size();
		memcpy(&parms->devTypeInfo, pRec, sizeof(FORE_DEV_TYPE));
		parms->pPubCli = this; 

		process_pubMsg((void*)parms);
	}
}

// bool _send_sub_data(void* context, int qos, FORE_DEV_DICT * pDevInfo, char* pServiceID, float val )
// { 
// 	if( context == NULL || pDevInfo == NULL ) return false;

// 	MQTTAsync  client = (MQTTAsync)context;
	
// 	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
// 	MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
	   
// 	int pos = 0;
// 	int SEND_MAX_REC = 50;  
// 	int idx = 0;

// 	char  topic[128] = {0};
// 	sprintf(topic, "%s/terminal/dataReport/%s/%s", "dataCenterApp", pDevInfo->deviceType, pDevInfo->dev_nodeID );

// 	int  token = 10000;
 
// 	token = getpid();

// 	rapidjson::StringBuffer s;
// 	rapidjson::Writer<rapidjson::StringBuffer> writer(s);
 	
// 	char  tmp[32] = {0}; 

// 	writer.StartObject();

// 	writer.Key("body");
// 	writer.StartObject();

// 	writer.Key("nodeId");
// 	sprintf(tmp, "%s", pDevInfo->dev_nodeID );
// 	writer.String(tmp);  

// 	writer.Key("val");
// 	sprintf(tmp, "%.2f", val);
// 	writer.String(tmp);  

// 	int  dataTime = 0;

// 	writer.EndObject();

// 	writer.Key("token");
// 	sprintf(tmp, "%d", token );
// 	writer.String(tmp);
// 	writer.Key("timestamp"); 

// 	// char tm[64];
// 	// LocalTime(tm);
// 	writer.String(base::toTimeString(dataTime).c_str());  
// 	// writer.String( tm );

// 	writer.EndObject(); 

// 	printf( "send payload: %s\n", s.GetString());
		 
// 	pubmsg.payload = (char*)s.GetString();
// 	pubmsg.payloadlen = s.GetSize() ;  
// 	pubmsg.retained = 0;
// 	pubmsg.qos = qos;
				
// 	opts.onSuccess =  onPubSend;
// 	opts.context = client;  

// 	base::my_printf( "send payload: %s\n", s.GetString());

//  	int rc = MQTTAsync_send(client, topic, pubmsg.payloadlen, pubmsg.payload,pubmsg.qos, pubmsg.retained, &opts);
// 	if ( rc  != MQTTASYNC_SUCCESS )
// 	{  
// 		base::my_printf( "Failed to  sendMessage  , return code %d\n", rc);
// 		return false;
// 	} 

// 	return true;
// }


// thread_return_type process_pubMsg(void* n)
void process_pubMsg(void* n)
{	
	int i = 0;
	int rc = 0;
	int wait_seconds = 30; 

	struct pub_thread_parms * pParm = (pub_thread_parms*)n;
	
	struct pub_thread_parms  parms; 
	parms.c = pParm->c;
	parms.index = pParm->index;
	parms.qos   = pParm->qos;
	parms.devTypeInfo  = pParm->devTypeInfo;
	parms.pPubCli = pParm->pPubCli;
	delete n;
	n = NULL;
	 
	base::MySleep(1000);
 
	// pthread_t   cur_tid = pthread_self();
	// base::my_printf( "Thread id=%lu, devtype=%s  begin.... \n", cur_tid,  parms.devTypeInfo.dev_type );

	// pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL); 
	// pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL); 

	std::map<string, std::vector<RT_DATA> >  mapNodeID2Rtdata; 

	std::map<string, ARYMEATEMPLATE>  tmp_mapServiceID2Meatempl; 
	TDbFun::query_meaTempl_by_devType(parms.devTypeInfo.dev_type, tmp_mapServiceID2Meatempl);
	// if( tmp_mapServiceID2Meatempl.size() <= 0 ) {
	// 	base::my_printf("THREAD id=%lu devtype=%s finished \n", cur_tid, parms.devTypeInfo.dev_type ); 
	// 	parms.pPubCli->CloseThreadHandle( parms.devTypeInfo.dev_type );
	// 	return NULL;
	// }

	std::map<string, ARYMEATEMPLATE>  mapServiceID2Meatempl; 

	for (std::map<string, ARYMEATEMPLATE>::iterator it = tmp_mapServiceID2Meatempl.begin(); it != tmp_mapServiceID2Meatempl.end();  ++it) 
	{   
		string			strServiceID			= (*it).first; 
		ARYMEATEMPLATE		tmp_ary				= (*it).second;  

		ARYMEATEMPLATE   ary;
		for( int i=0; i<tmp_ary.size(); i++ )
		{
			FORE_MEA_TEMPLATE & rec = tmp_ary[i];
			if( rec.zf_type == 0 )
				ary.push_back( rec );
		} 

		mapServiceID2Meatempl.insert( pair<string, ARYMEATEMPLATE>(strServiceID, ary) );
	}
	
	std::vector<FORE_DEV_DICT>   aryDev;
	TDbFun::GetDevs_ByDevType( parms.devTypeInfo.dev_type, aryDev ); 

	base::my_printf(" thread:  devType=%s  count=%d \n", parms.devTypeInfo.dev_type, aryDev.size() );
	

	for( int i=0; i<aryDev.size(); i++ )
	{
		FORE_DEV_DICT & dev = aryDev[i];
	 
		std::vector<RT_DATA>  rtAllData;
		printf("11111111111111111111\n");
		for (std::map<string, ARYMEATEMPLATE>::iterator it = mapServiceID2Meatempl.begin(); it != mapServiceID2Meatempl.end();  ++it) 
		{   
				string			strServiceID			= (*it).first; 
				ARYMEATEMPLATE		ary				= (*it).second;  
				// if( ary.size() <= 0 )  continue;
				
				std::vector<RT_DATA>  tmpRtAllData;
				TDbFun::load_rtdata( dev.dev_nodeID, ary, tmpRtAllData);  
				// if( tmpRtAllData.size() <= 0 ) continue;  

				printf("222222222222222222\n");
				pthread_mutex_lock(&server.mapmutex);
				for(int jj = 0; jj < ary.size(); ++jj)
				{
					auto iter = server.m_data.find(dev.dev_nodeID);
					printf("3333333333333333\n");
					// _send_sub_data(parms.c, parms.qos, &dev, (char*)strServiceID.c_str(), iter->second);  //0:all
				}
				pthread_mutex_unlock(&server.mapmutex);

				// rtAllData.insert(rtAllData.end(), tmpRtAllData.begin(), tmpRtAllData.end()); 
		}
		
		// if( rtAllData.size() > 0 ) {
		// 	mapNodeID2Rtdata[dev.dev_nodeID] = rtAllData ; 
		// }
	}

	return NULL;
}

int Pub_msgarrvd(void* context, char* topicName, int topicLen, MQTTAsync_message* m)
{
	CMQTTAsyncPubCli * ipCli = (CMQTTAsyncPubCli*)context; 
	 
	return 1;
}


#include "MD5.h"
pid_t  g_pid;


class CMQTTMsgSvr
{
public:
	CMQTTAsyncPubCli     pubcli; 

public:
	void start()
	{
		char mqttSvrIp[32] = "tcp://127.0.0.1:1883";
 
		pubcli.connect_mqtt(mqttSvrIp);
 
	} 
	void ReStart()
	{ 
		pubcli.ReStartPubThreasd( ); 
	}

	void check()
	{
		
	}
};


void CreateRtdb()
{
	EmptyAll(); 
 
	base::my_printf("CreateRtdb begin.....\n");
 
	CreateRTdb( "SYS_STRING_KEY_NAME",				SYS_STRING_KEY_NO,		SYS_STRING_KEY_COUNT,		sizeof(SYS_STRING_KEY));
	CreateRTdb( "FORE_DEV_TYPE_NAME",				FORE_DEV_TYPE_NO,		FORE_DEV_TYPE_COUNT,		sizeof(FORE_DEV_TYPE)); 
	CreateRTdb( "FORE_DEV_INFO_NAME",				FORE_DEV_INFO_NO,		FORE_DEV_INFO_COUNT,		sizeof(FORE_DEV_DICT));
	CreateRTdb( "FORE_MEA_TEMPLATE_NAME",			FORE_MEA_TEMPLATE_NO,	FORE_MEA_TEMPLATE_COUNT,	sizeof(FORE_MEA_TEMPLATE));
	
	CreateRTdb( "FORE_PARAM_CFG_NAME",				FORE_PARAM_CFG_NO,		FORE_PARAM_CFG_COUNT,		sizeof(FORE_PARAM_CFG)); 

	CreateRTdb( "FORE_YX_POINT_RTDB_NAME",			FORE_YX_POINT_NO,		FORE_YX_POINT_COUNT,		sizeof(FORE_YX_POINT));
	CreateRTdb( "FORE_YC_POINT_RTDB_NAME",			FORE_YC_POINT_NO,		FORE_YC_POINT_COUNT,		sizeof(FORE_YC_POINT)); 
	CreateRTdb( "FORE_YK_NO_NAME",					FORE_YK_NO,				FORE_YK_COUNT,				sizeof(FORE_YK));

	CreateRTdb( "FORE_EVENT_POINT_NAME",		FORE_EVENT_POINT_NO,	FORE_EVENT_POINT_COUNT,	sizeof(FORE_EVENT_POINT ));

	CreateRTdb( "FORE_CURVE_NAME",		FORE_CURVE_NO,	FORE_CURVE_COUNT,	sizeof(RTDB_YC_CURVE_5MIN));
	 	
	base::my_printf("CreateRtdb end.....\n");
}
 
 bool load_rtdb()
{
	if(  !base::is_exist_file("/rtdbms/LIB0002.data") ) { 
 
		if( !OpenDbms() ) {
			printf("msgSvr: open rtdbms error! \n"); 
			printf("rm -rf /rtdbms \n"); 
			system("rm -rf /rtdbms");
			return false;
		}
		else{
			printf("open rtdbms sucess!\n"); 
		}  
 
		void CreateRtdb();
		CreateRtdb(); 

		if( !TDbFun::set_rtdb_version(CURRENT_RTDB_VERSION) )
			printf("set_rtdb_version false !\n"); 

		SaveRtdb();
		
		base::my_printf("Load rtdb rtdbms\n");

		return true;
	} 
	else {  
		if( !OpenDbms() ) {
			base::my_printf("msgSvr: open rtdbms error! \n");  
			printf("rm -rf /rtdbms \n"); 
			system("rm -rf /rtdbms");
			return false;
		}
		else{
			base::my_printf("open rtdbms sucess!\n"); 
		}

		unsigned int ver = TDbFun::get_rtdb_Version();
		
		if( ver != CURRENT_RTDB_VERSION ) 
		{
			printf("rtdb ver[%d] != cur rtdb version[%d]!\n", ver, CURRENT_RTDB_VERSION ); 

			printf("rm -rf /rtdbms \n");
			system("rm -rf /rtdbms");

			return false;
		} 

		if( !LoadRtdb() ) 
		{
			base::my_printf(" Load rtdb rtdbms failed2! \n"); 
			return false;
		}
		else
		{
			base::my_printf(" Load rtdb rtdbms sucess2! \n"); 
		} 

		return true;
	}
}

void start_tcp()
{
	sleep(5);
	server.Bind();
    server.Listen();
    server.Run();
}


bool MY_PRINTF_TAG  = 0;
int main(int argc, char* argv[])
{
	signal(SIGKILL , exit_process);
	signal(SIGINT  , exit_process);
	signal(SIGTERM , exit_process);	 
	signal(SIGQUIT , exit_process);	 
	 
	if( argc == 2 ) { 
		if( strcmp(argv[1], "p") == 0 )
			MY_PRINTF_TAG = 1 ; 
	} 
	
	g_pid = getpid();
	readJson();
	thread serv = thread(start_tcp);
	thread mqttthread = thread(mqtt_thread);
	mqttthread.detach();

	base::MySleep(1000);

	printf( "\n mqtt pubilsh run successfully! pid =%d \n ", g_pid ); 
	  
	if( !load_rtdb() ) {
		base::MySleep(2000); 
		printf( " mqtt pubilsh exit \n " ); 
		exit(0);
	}
	
	base::MySleep(1000);
 

	CMQTTMsgSvr   mqtt_svr;
	// mqtt_svr.start();
 
	unsigned int  loop_count = 1;

	serv.join();

	// thread mqttthread = thread(mqtt_thread);
	// smysql.join();
	// dbthread.join();
	// mqttthread.detach();
   
	return 0;
}
 
  



// int main()
// {
// 	readJson();
// 	// thread smysql = thread(save_mysql);
// 	// thread dbthread = thread(db_thread);

// 	thread mqttthread = thread(mqtt_thread);
// 	// smysql.join();
// 	// dbthread.join();
// 	mqttthread.detach();

//     server.Bind();
//     server.Listen();
//     server.Run();
//     return 0;
// }
